
#include <string>
#include <vector>
#include <map>
#include <google/protobuf/service.h>
#include <google/protobuf/descriptor.h>
#include <google/protobuf/message.h>
#include <google/protobuf/stubs/common.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include "utility.h"
#include "meta.h"
#include "rpcserverevent.h"
#include "eventpoller.h"
#include "socket_utility.h"

using std::string;
using std::vector;
using std::map;

namespace eventrpc {
  struct RpcMethod {
   public:
    RpcMethod(gpb::Service *service,
              const gpb::Message *request,
              const gpb::Message *response,
              const gpb::MethodDescriptor *method)
      : service_(service)
        , request_(request)
        , response_(response)
        , method_(method) {
        }

    gpb::Service *service_;
    const gpb::Message *request_;
    const gpb::Message *response_;
    const gpb::MethodDescriptor *method_;
  };

  typedef map<uint32_t, RpcMethod*> RpcMethodMap;

  struct ClientEvent : public Event {
   public:
    ClientEvent(int fd, const RpcMethodMap &rpc_methods,
                RpcServerEvent *server_event)
      : Event(READ_EVENT, fd)
      , rpc_methods_(rpc_methods)
      , server_event_(server_event) {
    }

    virtual int OnWrite();

    virtual int OnRead();

    void HandleServiceDone();

   private:
    char buffer_[100];
    RpcMethodMap rpc_methods_;
    Meta meta_;
    RpcServerEvent *server_event_;
    gpb::Message *request_;
    gpb::Message *response_;
    const gpb::MethodDescriptor *method_;
    string message_;
  };

  int ClientEvent::OnWrite() {
    printf("OnWrite\n");

    //write(fd_, buffer_, strlen(buffer_));
    write(fd_, message_.c_str(), message_.length());

    close(fd_);
    event_poller_->DelEvent(event_, this);

    return 0;
  }

  int ClientEvent::OnRead() {
    size_t len = read(fd_, buffer_, META_LEN);
    buffer_[len] = '\0';
    meta_.Encode(buffer_);
    printf("meta len: %d, method id: %d, message len: %d\n",
           META_LEN, meta_.method_id(), meta_.message_len());

    RpcMethodMap::iterator iter;
    if ((iter = rpc_methods_.find(meta_.method_id())) != rpc_methods_.end()) {
      printf("method name: %s\n",
             iter->second->method_->full_name().c_str());
    } else {
      printf("unfind id: %d\n", meta_.method_id());
      return -1;
    }
    RpcMethod *rpc_method = iter->second;
    len = read(fd_, buffer_, meta_.message_len());
    printf("len: %d, read: %s\n", len, buffer_);
    method_ = rpc_method->method_;;
    request_ = rpc_method->request_->New();
    response_ = rpc_method->response_->New();

    request_->ParseFromString(string(buffer_));

    gpb::Closure *done = gpb::NewCallback(this,
                                          &ClientEvent::HandleServiceDone);
    rpc_method->service_->CallMethod(method_,
                                     NULL,
                                     request_, response_, done);
    //close(client_event_->fd_);
    //event_poller_->DelEvent(client_event_->event_, client_event_);

    return 0;
  }

  void ClientEvent::HandleServiceDone() {
    printf("in HandleServiceDone\n");
    meta_.set_method_id(method_->full_name());
    meta_.set_message_len(response_->ByteSize());
    message_.append(meta_.Decode(), META_LEN);
    response_->AppendToString(&message_);
    event_poller_->AddEvent(WRITE_EVENT, this);
    delete request_;
    delete response_;
  }

  struct RpcServerEvent::Impl {
   public:
    Impl(const char *ip, int port, RpcServerEvent *serverevent);

    ~Impl();

    int OnWrite();

    int OnRead();

    bool RegisterService(gpb::Service *service);

   private:
    RpcServerEvent *server_event_;
    vector<ClientEvent*> client_events_;
    RpcMethodMap rpc_methods_;
  };

  RpcServerEvent::Impl::Impl(const char *ip, int port,
                             RpcServerEvent *server_event)
    : server_event_(server_event) {
      server_event_->fd_ = Listen(ip, port);
  }

  RpcServerEvent::Impl::~Impl() {
  }

  int RpcServerEvent::Impl::OnWrite() {
    return -1;
  }

  int RpcServerEvent::Impl::OnRead() {
    int fd;
    struct sockaddr_in addr;
    socklen_t len = sizeof(addr);

    fd = accept(server_event_->fd_, (struct sockaddr *)&addr, &len);
    if (fd == -1) {
      return -1;
    }

    if (SetNonBlocking(fd) < 0) {
      return -1;
    }

    ClientEvent *client_event = new ClientEvent(
        fd, rpc_methods_, server_event_);
    server_event_->event_poller()->AddEvent(READ_EVENT, client_event);
    client_events_.push_back(client_event);

    return 0;
  }

  bool RpcServerEvent::Impl::RegisterService(gpb::Service *service) {
    const gpb::ServiceDescriptor *descriptor = service->GetDescriptor();
    for (int i = 0; i < descriptor->method_count(); ++i) {
      const gpb::MethodDescriptor *method = descriptor->method(i);
      const google::protobuf::Message *request =
        &service->GetRequestPrototype(method);
      const google::protobuf::Message *response =
        &service->GetResponsePrototype(method);
      RpcMethod *rpc_method = new RpcMethod(service, request,
                                            response, method);

      uint32_t method_id = hash_string(method->full_name());
      printf("register method %s for id %d\n",
             method->full_name().c_str(),
             method_id);
      rpc_methods_[method_id] = rpc_method;
    }

    return true;
  }

  RpcServerEvent::RpcServerEvent(const char *ip, int port)
    : Event(READ_EVENT)
      , impl_(new Impl(ip, port, this)) {
      }

  RpcServerEvent::~RpcServerEvent() {
    delete impl_;
  }

  int RpcServerEvent::OnRead() {
    return impl_->OnRead();
  }

  int RpcServerEvent::OnWrite() {
    return impl_->OnWrite();
  }

  bool RpcServerEvent::RegisterService(gpb::Service* service) {
    return impl_->RegisterService(service);
  }
};
